#include "StdAfx.h"
#include "RowWindowOperator.h"
#include "WindowSynopsis.h"
#include "QueueEntity.h"
RowWindowOperator::RowWindowOperator(int rowWindowSize):rowWindowSize(rowWindowSize)
{
	this->windowSynopsis.reset(new WindowSynopsis());
	this->currentRowNumberInSynopsis = 0;
	this->lastInputTimestamp = 0;
}


RowWindowOperator::~RowWindowOperator(void)
{
}


void RowWindowOperator::execution(int executionTimes)
{
#ifdef DEBUG
	std::cout<<"===================operator begin================="<<std::endl;
	std::cout<<"operatorid : "<<this->getId()<<std::endl;
#endif
	assert(this->getInputQueueList().size()==1);
	assert(this->getOutputQueueList().size()==1);
	boost::shared_ptr<QueueEntity>inputQueue = this->getInputQueueList().front();
	boost::shared_ptr<QueueEntity>outputQueue = this->getOutputQueueList().front();
	if(this->getOperatorState() == BLOCKED)
	{
		dealWithBlock(); // if we get into block state in the last execution, we should deal with that first
	}
	if(this->getOperatorState() == BLOCKED)// if we are blocked currently  
	{
		return;
#ifdef DEBUG
		std::cout<<"===================operator over================="<<std::endl;
#endif
	}

	for(int i = 0; i < executionTimes ; i++)
	{
		Element inputElement;
		
		if(inputQueue->isEmpty())   // the input queue is not empty
		{
			break;
		}
		if(outputQueue->isFull())   // the output queue is not full 
		{
			break;
		}
		inputQueue->dequeue(inputElement);
		

		assert(inputElement.mark==PLUS_MARK);
		assert(lastInputTimestamp<= inputElement.timestamp);
		lastInputTimestamp = inputElement.timestamp;

		if(this->windowSynopsis->isFull())   // the window synopsis is not full
		{
			throw std::runtime_error("synopsis full");
		}
		
		this->windowSynopsis->insertElement(inputElement);
		this->currentRowNumberInSynopsis++;

		outputQueue->enqueue(inputElement);//forward the same element

		Element outputElement;
		if(this->currentRowNumberInSynopsis>this->rowWindowSize)
		{  // generate oldest element out
			this->windowSynopsis->getOldestElement(outputElement);
			this->windowSynopsis->deleteOldestElement();
			this->currentRowNumberInSynopsis--;
			outputElement.mark = MINUS_MARK;
			outputElement.timestamp = inputElement.timestamp;
		
			//if output queue is full, change the state to blocked and buffer the element;
			if(outputQueue->isFull())
			{
				this->setOperatorState(BLOCKED);
				this->blockedElement = outputElement;
				return;
			}
			else
			{
				outputQueue->enqueue(outputElement);

			}
		}
	}
#ifdef DEBUG
	std::cout<<"===================operator over================="<<std::endl;
#endif
}
void RowWindowOperator::dealWithBlock(void)
{
#ifdef DEBUG
	std::cout<<"**********deal with block begin************"<<std::endl;
#endif
	
	
	assert(this->getOperatorState() == BLOCKED);
	boost::shared_ptr<QueueEntity>outputQueue = this->getOutputQueueList().front();
	outputQueue->enqueue(this->blockedElement);
	this->setOperatorState(NORMAL);
#ifdef DEBUG
	std::cout<<"**********deal with block over************"<<std::endl;
#endif
}